import datetime
import logging

from absl import app

from coin.base.datetime_util import to_datetime

from experimental.prophet.graph import graph
from experimental.prophet.graph.shape import Shape

from experimental.prophet.graph_runner.dataframe import run_from_dataframe
from experimental.prophet.graph_runner.fastfeed import run_from_fastfeed_multiprocess

from experimental.prophet.ops import globals
from experimental.prophet.ops.constant import constant
from experimental.prophet.ops.placeholder import placeholder
from experimental.prophet.ops.fastfeed import fastfeed_coin
from experimental.prophet.ops import timeseries
from experimental.prophet.ops import y_gen
from experimental.prophet.ops.control_flow import cond
from experimental.prophet.ops.elemwise_math import *
from experimental.prophet.ops.aggregator import scalar_value_aggregator


def model():
  ts = globals.global_timestamp()
  xbtusd = fastfeed_coin('BTC-USD.PERPETUAL.Bitmex.XBTUSD',
                         'Futures.Bitmex',
                         'feed-01.ap-northeast-1.aws')
  n1 = graph.create_node('fastfeed.BookBestLevel', [xbtusd.book])
  xbtusd_ask0_p = n1.outputs[1].named_ref('xbtusd_ask0_p')
  xbtusd_bid0_p = n1.outputs[3].named_ref('xbtusd_bid0_p')

  n2 = graph.create_node('fastfeed.Trade', [xbtusd.trade])
  xbtusd_tp = n2.outputs[1].named_ref('xbtusd_tp')
  xbtusd_tq = n2.outputs[2].named_ref('xbtusd_tq')
  xbtusd_tside = n2.outputs[3].named_ref('xbtusd_tside')
  xbtudf_tp_shift = timeseries.shift(xbtusd_tp, 1)

  xbtusd_tp_mw10s = timeseries.time_moving_window(xbtusd_tp, '10s')
  xbtusd_tp_mstd10s = xbtusd_tp_mw10s.std()
  xbtusd_tp_mmin10s = xbtusd_tp_mw10s.min()
  xbtusd_tp_mmin10s = xbtusd_tp_mw10s.max()
  xbtusd_tp_mmaxdiff10s = xbtusd_tp_mw10s.max_min_diff()

  xbtusd_tq_msum10s = timeseries.time_moving_window(xbtusd_tq, '10s').sum()
  xbtusd_buyq_msum10s = timeseries.time_moving_window(
      cond(xbtusd_tside == 1, xbtusd_tq, 0.).named_ref('xbtusd_buyq'), '10s').sum()
  xbtusd_sellq_msum10s = timeseries.time_moving_window(
      cond(xbtusd_tside == 2, xbtusd_tq, 0.).named_ref('xbtusd_sellq'), '10s').sum()

  windows = ['1s', '2s', '3s', '5s', '10s', '1m', '2m', '5m', '10m']
  xbtusd_p_pasts = timeseries.time_shift([xbtusd_ask0_p, xbtusd_bid0_p], windows)

  xbtusd_mid = (xbtusd_ask0_p + xbtusd_bid0_p) / 2
  xbtusd_mid_10s = timeseries.time_shift(xbtusd_mid, '10s')
  mid_ret = (xbtusd_mid - xbtusd_mid_10s).named_ref('mid_ret')
  y_mid_10s = y_gen.time_shift_y(mid_ret, '10s')

  with graph.control_if(ts % 10**9 == 0):
    aggregator = scalar_value_aggregator([
        ts,
        y_mid_10s,
        mid_ret,
        xbtusd_tp_mmaxdiff10s,
        xbtusd_tp_mstd10s,
        xbtusd_tq_msum10s,
        xbtusd_buyq_msum10s,
        xbtusd_sellq_msum10s,
        xbtusd_ask0_p,
        xbtusd_bid0_p,
        *xbtusd_p_pasts
    ])
  return aggregator


def main(argv):
  df = run_from_fastfeed_multiprocess(model,
                                      datetime.date(2019, 4, 1),
                                      datetime.date(2019, 4, 2),
                                      machine='feed-01.ap-northeast-1.aws',
                                      periodic_eval=datetime.timedelta(milliseconds=100),
                                      max_workers=None,
                                      use_run_cache=True,
                                      gen_dataframe_output=True)
  print(df)


if __name__ == '__main__':
  logging.basicConfig(
      level='DEBUG',
      format='%(levelname)8s %(asctime)s %(name)s %(filename)s:%(lineno)d] %(message)s')
  logging.getLogger('coin.exchange.util.feed_checker').setLevel(logging.WARNING)
  app.run(main)
'''
ethusd = fastfeed_coin('ETH-USD.PERPETUAL.Bitmex.ETHUSD', 'Futures.Bitmex', 'feed-01.ap-northeast-1.aws')
n2 = graph.create_node('BookBestPrice', [ethusd.book])
ethusd_ask0_p = n2.outputs[0].named_ref('ethusd_ask0_p')
ethusd_bid0_p = n2.outputs[2].named_ref('ethusd_bid0_p')

#with graph.control_if(fmod(xbtusd_ask0_p, 1) == 0):
#with graph.control_if(ts % 10**9 == 0):
df = run_from_fastfeed(
      [ts, y_mid_10s, mid_ret, xbtusd_tp_mmaxdiff10s, xbtusd_tp_mstd10s, xbtusd_tq_msum10s, xbtusd_buyq_msum10s, xbtusd_sellq_msum10s, xbtusd_ask0_p, xbtusd_bid0_p, *xbtusd_p_pasts],
       datetime.date(2019, 5, 2), datetime.date(2019, 5, 3),
       machine='feed-01.ap-northeast-1.aws',
       periodic_eval=datetime.timedelta(milliseconds=100))

print('Shape:', df.shape)
print('Columns:', ', '.join(df.columns))
df.to_pickle('dump_1.pkl.gz')
'''
